package a.b.c;

import biz.ws.ws.Message;
import biz.ws.ws.client.WsClient;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * @创建人 wsdc
 * @时间 2021/4/1
 * @描述 $
 */
public class WsClientTest0 {
    @Test
    public void testConnect() throws Exception {
        Semaphore sem = new Semaphore(0);
        WsClient client = new WsClient(false, (ctx, frame) -> {
            //  收到消息
            //ctx.channel().close();
            String text = ((TextWebSocketFrame) frame).text();
            System.out.println(text);
            sem.release();
        });


        client.connect("ws://127.0.0.1:7777/ws/ksv.do?userId=1&loginFrom=window", future -> {
            //  链接成功之后的回调
            //  发送一条消息
            future.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Message.beat(1l))))
                    .addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {

                        }
                    });
        });

        sem.acquire();
    }

    /**
     *  AB对话测试
     */
    @Test
    public void chatTest() throws Exception{
        WsClient client = new WsClient(false, (ctx, frame) -> {
            //  收到消息
            //ctx.channel().close();
            String text = ((TextWebSocketFrame) frame).text();
            System.out.println(text);
        });


//        client.connect("ws://127.0.0.1:7777/ws/ksv.do?userId=2&loginFrom=window", future -> {
//            //  链接成功之后的回调
//            //  发送一条消息
//            Flux.interval(Duration.ofSeconds(1))
//                    .subscribe(k -> {
//                        future.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Message.sendTo(2l,2l,"forLove to 2"))))
//                                .addListener(new ChannelFutureListener() {
//                                    @Override
//                                    public void operationComplete(ChannelFuture future) throws Exception {
//
//                                    }
//                                });
//                    });
//
//        });

        client.connect("ws://127.0.0.1:7777/ws/ksv.do?userId=2&loginFrom=window", future -> {

        });

        LockSupport.park();
    }

    /**
     * 测试
     * <li>    10
     * <li>    100
     * <li>    1000
     * <li>    5000
     * <li>    10000
     * <li>    20000
     * <li>    40000
     * <li>    80000
     * <li>    100000
     * 个链接系统的反应
     * <p>
     * 发送心跳 查看心跳的耗时
     */
    @Test
    public void connectAndBeatConcurrent() throws Exception {
        int max = 10000;

        Map<Integer, DataInfo> infoMap = new HashMap<>();
        Semaphore semaphore = new Semaphore(0);

        AtomicInteger ai = new AtomicInteger(0);

        long vv = System.currentTimeMillis();
        WsClient client = new WsClient(false, (ctx, frame) -> {
            //  收到消息
            DataInfo dataInfo = infoMap.get(ctx.channel().hashCode());
            dataInfo.receiveBeatTime = LocalDateTime.now();

            semaphore.release();
            System.out.println(ai.incrementAndGet());
        });

        for (int i = 0; i < max; i++) {
            DataInfo info = new DataInfo();
            info.id = i;
            info.createTime = LocalDateTime.now();

            long s = System.currentTimeMillis();
            client.connect(String.format("ws://192.168.31.99:7777/ws/ksv.do?userId=%d&loginFrom=window", i), future -> {

                infoMap.put(future.channel().hashCode(), info);
                info.connectSuccessTime = LocalDateTime.now();
                info.future = future;

                //  一个随机时间分发数据
                Mono.delay(Duration.ofSeconds(RandomUtil.randomInt(15)))
                        .subscribe(k -> {
                            info.sendBeatTime = LocalDateTime.now();
                            future.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(Message.beat(1l))))
                                    .addListener(new ChannelFutureListener() {
                                        @Override
                                        public void operationComplete(ChannelFuture future) throws Exception {

                                        }
                                    });
                        });

                //  链接成功之后的回调
                //  发送一条消息

            });
            long e = System.currentTimeMillis();
            System.out.println("time = "+(e-s));
            s = e;
        }

        semaphore.acquire(max);
        int connectTime = 0;
        int beatTime = 0;
        int connectTimeMax = 0;
        int beatTimeMax = 0;
        for (DataInfo value : infoMap.values()) {
            System.out.println(value.toString());
            Mono.delay(Duration.ofSeconds(3))
                    .subscribe(k -> {
                        value.future.channel().close();
                    });
            int v1 = (int) (Duration.between(value.createTime, value.connectSuccessTime).getNano() / 1000 / 1000l);
            int v2 = (int) (Duration.between(value.sendBeatTime, value.receiveBeatTime).getNano() / 1000 / 1000l);

            connectTime += v1;
            beatTime += v2;

            connectTimeMax = connectTimeMax < v1 ? v1 : connectTimeMax;
            beatTimeMax = beatTimeMax < v2 ? v2 : beatTimeMax;


        }

        System.out.println(connectTime / max);
        System.out.println(beatTime / max);

        System.out.println(connectTimeMax+" 最大链接时长");
        System.out.println(beatTimeMax+"    最大心跳时长");


        System.out.println("总计耗时 "+(System.currentTimeMillis()-vv));
        LockSupport.park();
    }

    public static class DataInfo {
        private int id;
        private ChannelFuture future;
        private LocalDateTime createTime;
        private LocalDateTime connectSuccessTime;
        private LocalDateTime sendBeatTime;
        private LocalDateTime receiveBeatTime;

        @Override
        public String toString() {
            return "id = " + id + " connectTime = " + Duration.between(createTime, connectSuccessTime).getNano() / 1000 / 1000l
                    + " beat time = " + Duration.between(sendBeatTime, receiveBeatTime).getNano() / 1000l / 1000l;
        }
    }
}
